package com.atguigu.app.dwd.db;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.KafkaUtil;
import com.ibm.icu.util.Output;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author hfh
 * @version 1.0
 * @date 2022/9/3 18:55
 */
public class DwdUserRegister {
    public static void main(String[] args) throws Exception {
        //获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //设置状态后端
        /*env.enableCheckpointing(5*60*1000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setCheckpointTimeout(3*60*1000L);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        //TODO 3 从kafka topic_db中读取数据
        String topicName = "topic_db";
        String groupId = "dwd_user_register";
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));

        //TODO 4 过滤出用户表user_info 新增用户数据并转化为JSONObject
        OutputTag<String> dirty = new OutputTag<String>("dirty"){};
        SingleOutputStreamOperator<String> process = streamSource.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    String table = jsonObject.getString("table");
                    String type = jsonObject.getString("type");
                    long ts = jsonObject.getLong("ts") * 1000L;
                    jsonObject.put("ts",ts);
                    if ("user_info".equals(table) && "insert".equals(type)) {
                        out.collect(jsonObject.toJSONString());
                    }
                } catch (Exception e) {
                    ctx.output(dirty, value);
                }
            }
        });
        /**
         * {
         *     "birthday": "1975-08-08",
         *     "create_time": "2022-03-08 00:00:00",
         *     "login_name": "z2w9azxh",
         *     "user_level": "1",
         *     "phone_num": "13923429898",
         *     "id": 1950,
         *     "email": "z2w9azxh@163.com"
         * }
         */
        String topicName1 = "dwd_user_register";
        process.addSink(KafkaUtil.getKafkaProducer(topicName1));

        env.execute(groupId);


    }
}
